package com.atguigu.process;

import com.atguigu.bean.WaterSensor;
import com.atguigu.function.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class KeyedProcessTimerDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> stream1 = env
                .socketTextStream("10.75.186.206", 9999)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((value, ts) -> value.getTs() * 1000L)
                );

        /**
         * TODO 定时器
         * 1、keyed才有
         * 2、事件时间定时器，通过watermark来触发的
         *    watermark >= 注册的时间
         *    注意： watermark = 当前最大事件时间 - 等待时间 -1ms， 因为 -1ms，所以会推迟一条数据
         *        比如， 5s的定时器，
         *        如果 等待=3s， watermark = 8s - 3s -1ms = 4999ms,不会触发5s的定时器
         *        需要 watermark = 9s -3s -1ms = 5999ms ，才能去触发 5s的定时器
         * 3、在process中获取当前watermark，显示的是上一次的watermark
         *    =》因为process还没接收到这条数据对应生成的新watermark
         */
        stream1.keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        Long currentEventTime = ctx.timestamp(); // 数据中提取出来的事件时间

                        // 注册定时器： 处理时间、事件时间
                        // 参数是希望触发的时间戳  而不是多长时间以后
//                        ctx.timerService().registerEventTimeTimer(currentEventTime + 5000);
                        ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 3000);
                        System.out.println("当前key=" + ctx.getCurrentKey() + ",当前时间=" + currentEventTime + ",注册了一个5s的定时器");
                    }

                    // * TODO 2.时间进展到定时器注册的时间，调用该方法
                    // 如果是事件时间  则在注册时间 + 水位线偏移量 + 1ms 的时候触发
                    // 如果是执行时间  则在 注册时刻 的时候触发
                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
//                        super.onTimer(timestamp, ctx, out);
                        String currentKey = ctx.getCurrentKey();
                        System.out.println("key=" + currentKey + "现在时间是" + timestamp + "定时器触发");
                    }
                });

        env.execute();
    }
}
